Bun.serve: never terminate a failed body stream as a complete message - #40596

…(error) (#40669)

### Problem
- A `ReadableStream` pumped into a native sink fails, and the sink
treats it as a clean end. `S3Client.write(key, stream)` commits the
buffered bytes as a complete single-file PUT (the object exists,
truncated) while the returned promise rejects with the source error.
`new HTMLRewriter().transform(new Response(stream)).text()` resolves to
`""` when the stream is already errored at `transform()` time.
- Cause: `readStreamIntoSink` calls `controller.close(error)` on the
sink controller before it rejects the pump promise
(`BunStreamSource.cpp`, `rsisAbrupt` → `rsisSinkClose`). The generated
`${controller}__close` (`src/codegen/generate-jssink.ts`) never read the
argument, and `JSSink::js_close` (`src/runtime/webcore/Sink.rs`) always
called `end(None)`. Every native sink saw EOF.

### Fix
- `close(reason)` now reaches the sink: the generated controller method
passes `callFrame->argument(0)` to `${name}__close`, and `js_close`
routes a non-empty reason to a new `JsSinkType::close_with_error(this,
global, reason)`. Its default is the old clean end, so
`ArrayBufferSink`, `FetchRequestBodySink`, the HTTP response sinks and
`FileSink` behave as before (their owners already handle the pump
rejection).
- `NetworkSink` (S3) fails the upload instead of committing: `task.fail`
before any PUT is scheduled, with the source error stashed so the
caller's promise rejects with it. `RewriterPipe` fails the rewrite with
the error, so the output body rejects.
- Correct because the pump has already decided the body is incomplete
when it closes with an error; the sink is the only party that did not
know. The pump rejection reactions still run afterwards and still
release their refs, so the ownership protocol of each sink is unchanged.
- Verified: `test/js/bun/s3/s3.test.ts` (two new rows, stock bun commits
a 2048 byte PUT) and `test/js/workerd/html-rewriter.test.js` (one new
test, stock bun resolves `""`). Also `filesink.test.ts`, the
`spawn-stdin-*` files, `streams.test.js`,
`serve-stream-body-error.test.ts`,
`serve-direct-readable-stream.test.ts`, `body.test.ts`,
`body-stream.test.ts`, `fetch.stream.test.ts`,
`fetch-backpressure.test.ts`, `html-rewriter-doctype.test.ts`.

### Background
- A native sink (`JsSinkType` in `Sink.rs`) is a Rust writer behind a JS
controller object. `JSSink::assign_to_stream` starts
`readStreamIntoSink`, the JS pump: it reads the stream and calls the
controller's `write`, `end`, or `close(error)`. The pump also returns a
promise the sink's owner reacts to.
- `close_with_error` takes `*mut Self`, like `finalize`:
`NetworkSink::fail` fires the S3 task's completion callback
synchronously, and that callback borrows the sink again.
- The sink's own JS `close()` method (`${name}__doClose`) still passes
no reason.

<details><summary>Notes</summary>

Wire before the fix, S3 stub on loopback, stream enqueues two 1 KB
chunks then `controller.error(new Error("boom"))`: the stub receives
`PUT /my_bucket/obj` with `Content-Length: 2048` and a complete body;
`client.write` rejects with `boom`. After: no PUT, the same rejection. A
stream that closes cleanly still produces the PUT.

HTMLRewriter before the fix: a stream that errors after the first chunk
already rejected (the pump's reject reaction is the terminal step on the
Pending arm). Only the already-errored input resolved to `""`:
`readMany` throws synchronously, so `close(error)` and the rejection
both happen inside `transform()`, the pipe had already run `end_rewrite`
from the clean close, and the Rejected arm's
`end_from_stream(Some(err))` found it done.

FileSink (spawn stdin, `Bun.write(path, stream)`) keeps the default. Its
pump reactions already handle the failure (`Bun.write` rejects; the
spawn path aborts the stream). Closing without flushing would drop the
bytes the source produced before it failed depending on whether the
auto-flusher had run yet, and
`spawn-stdin-readable-stream-edge-cases.test.ts` ("ReadableStream with
exception in pull") pins that those bytes reach the child. A pipe has no
incomplete-message signal beyond EOF, so there is nothing else for the
sink to do there; what a JS observer for a failed spawn stdin stream
should look like is a separate question.

Related: #40596 fixes the same family on the `Bun.serve` response side.

</details>

<!-- robobun:evidence:begin -->

---

**no test proof** · iteration 7 · platform-specific test(s) that do not
run on this machine, deferring to CI, which covers all platforms:
test/js/workerd/html-rewriter.test.js

<!-- robobun:evidence:end -->

…with the buffer (#38003)

### Problem

- A native byte stream first read after its producer failed ends clean
after 0 bytes. Sources: `fetch` body, `req.body`, `S3File.stream()`,
`HTMLRewriter` output. `Bun.serve` answers an empty `200`. An S3 copy
stores an empty object.
- If a reader took part of the buffer, the process aborts: `panic: range
end index 262144 out of range for slice of length 0`
(`ByteStream::take_buffer`).
- Cause: `ByteStream::append(Err)` stores the error, drops `buffer` and
keeps `offset`. `on_start`, `to_any_blob` and `Bun.serve` read that as
"ended, empty".

### Fix

- A stored error makes `on_start` return `Start::Ready` and
`to_any_blob` return `None`, so the first pull delivers it. `Bun.serve`
hands it to its sink, and `end_chunk` calls `error()`.
- `buffer` and `offset` become one `Buffered` value. Every method that
empties the bytes resets the index.
- Verified: six test files (Notes). 24 new tests fail without the
change.
- Self-reviewed: 2 concerns raised, 2 addressed.

### Background

- `ByteStream` is the native source of a streamed body.
- A pull copies into a view. Bytes that do not fit stay buffered.
`offset` counts the ones already taken.
- Weighed a store in `append(Err)` and a clamp in `take_buffer`. Six
sites empty the buffer, and a clamp hides a wrong index.

### Downsides

- A handler that returned such a body got a silent `200`. It now gets
`error()` or the default 500.
- One test of `pending.result` per stream start and per `Bun.serve`
attach. Release `text`: +768 bytes.
- Still open: `Bun.write(path, stream)` rejects after it truncated the
file. A started body that did not fail loses its queued bytes (#33461,
#42281).

<details><summary>Notes</summary>

**Origin and reach.** No user reported either symptom. A fuzzer found
the panic, and a differential run against node v26.3.0 found the empty
body. The panic ships since 1.4.0: #38656 gave `take_buffer` its
`drain(..consumed)`. The review thread at
#43410 (comment) named
the stale `offset` on 09-19 and got the answer that the crash would have
its own change with its own test. This is that change, inside this PR so
that `ByteStream.rs` has one pull.

**Three routes reach `take_buffer` with an index left behind an emptied
buffer.**

1. `Bun.serve` renders a `Source::Bytes` body that a consumer started
and let go (`res.body.getReader().releaseLock()`), after the producer
failed. `do_render_with_body` calls `drain()`.
2. `Readable.fromWeb()` on such a stream. `on_start` took the buffer and
kept the index, then `native-readable.ts` calls `drain()`. This one
needs no failure.
3. The stream's own pull loop in text mode (`res.textStream()`). The
adapter keeps its pull view there, so `nativeSourcePullImpl`
(`BunStreamSource.cpp`) calls `handle.drain()` before each pull. A body
that fails after a partial read is enough. No second consumer is
involved.

Sites that emptied `buffer` and kept `offset` on main: `append(Err)`,
the buffer-action error arm of `on_data`, `on_cancel`, the stored-error
arm of `to_buffered_value`, the `OwnedAndDone` arm of `on_start`,
`to_any_blob`. `Buffered` offers `take`, `clear`, `copy_out`, `adopt`
and `extend`, and nothing else writes the two fields.

**Before and after** (release build of main, debug build of this
branch):

| script | main | this PR |
| --- | --- | --- |
| `HTMLRewriter` output, probed with `getReader().releaseLock()`,
400,007 bytes of output, then a handler throws, returned from
`Bun.serve` | panic, exit 134 | `500`, `error()` got `boom` |
| the same with exactly one pull view of output (262,144 bytes) | `200
""` | `500`, `error()` got `boom` |
| the same after only `void res.body` | `200 ""` | `500`, `error()` got
`boom` |
| `const body = res.body`, upstream resets, `for await (const c of
body)` | ends cleanly after 0 of 100,000 bytes | rejects `ECONNRESET`
(node: rejects) |
| `res.textStream()` reader, one view read, upstream resets, next read |
panic, exit 134 | the read rejects `ECONNRESET` |
| `Readable.fromWeb()` on a probed rewriter output, clean end | panic,
exit 134 | the bytes the native source still holds |
| `req.body` read after the uploading client went away | `{ done: true
}` | rejects `AbortError` |
| `new Response(body).arrayBuffer()` / `bytes()` / `blob()` / `json()`
on a body that failed while idle | resolve empty | reject `ECONNRESET` |
| `new Bun.S3Client({}).file(key).stream()` (no credentials), then `for
await`, a reader, `pipeTo`, `tee`, `new Response(s).bytes()` or
`.blob()` | end cleanly after 0 bytes | reject
`ERR_S3_MISSING_CREDENTIALS` |
| `client.file(key).stream()` read after the download failed: a body cut
short, a missing key (404), a denied key (403) | ends cleanly after 0
bytes | rejects `ConnectionClosed`, `NoSuchKey`, `AccessDenied` |
| `client.write(key, new Response(s))` or `Bun.write(s3file, new
Response(s))` with such a stream `s` | resolves 0, a `PUT` of 0 bytes
replaces the object | rejects, no request goes out |
| `fetch(url, { method: "POST", body: s })` | resolves 200, the server
gets a complete POST with `Content-Length: 0` | rejects, the server gets
no complete request |
| `HTMLRewriter` output `.body` taken, a handler throws or the upstream
resets, then the first read | `{ done: true }` | rejects with the
handler's or the upstream's error |

**Four sources, one script.** Each transfer announces 100 bytes, sends
50, and is cut by a FIN 20 ms later. The stream is taken at once and
read 100 ms later: `(await fetch(url)).body`, `req.body` in `Bun.serve`,
`s3.file(key).stream()`, `new HTMLRewriter().transform(await
fetch(url)).body`. A release canary (367d939) and a debug build of
main bd599f5 end all four cleanly with 0 bytes, 3 of 3 runs each. A
debug build of this branch rejects all four, 3 of 3 runs, with the error
the same read gets when it is made at once: `ECONNRESET`, `AbortError`,
`S3Error ConnectionClosed`, `ECONNRESET`.

**What stays open.** For a stream source, `Bun.write(path, source)`
opens the destination with `O_TRUNC` before it reads the stream
(`pipe_readable_stream_to_blob`, `Blob.rs:1448`). With a failed stream,
or a `Response` or `Request` that wraps one, the write now rejects, and
a file that held 777 bytes holds 0. Main does the same for a bare
stream, and resolved 0 for the wrappers. Only `Bun.write(path, res)`
with the fetch `Response` itself keeps the old bytes. An S3 destination
keeps the old object, because no request goes out. This PR does not
change when `Bun.write` truncates.

Two behaviours change only in the state that used to abort: `on_start`
and `resume` hand over the bytes no consumer took, not the whole buffer,
so a prefix a reader already holds is not sent twice. The `to_write ==
0` arm of `on_pull` is gone: it handled a buffer whose bytes were all
taken but not dropped, which cannot exist now.

The `HTMLRewriter` paces its input by the whole buffer, the taken prefix
included (#38656). `Buffered::held_len()` keeps that. Counting only the
untaken bytes made `html-rewriter.test.js` "a locked reader paces the
input by its reads" fail, which is how the difference showed.

**Self-review.** Two concerns survived: that the change may claim
severity but not demand (so this body states that no user reported it
and lists the routes), and that a rename in `ByteStream.rs` races the
other open pulls on the file (it now ships inside this one).

**Since the first revision of this PR.** Merged up to main. The
`Blob.rs` hunk is gone: main rewrote `pipe_readable_stream_to_blob` and
marks the pump's rejected promise handled there.
`s3-write-to-file-sync-close.test.ts` now asserts the rejection instead
of accepting both outcomes. `end_sink_with_pending_error` uses main's
`detach_sink` plus `sink.end`. The fixture variants sit in main's
`nativeBodies` table (#40596). The "failed after attach" control is
dropped: main's `proxied-fetch-after-status` row covers it with #40596's
contract (the connection closes without a complete message). One variant
is new: the probed rewriter output from the first row above.

**Tests.** `bun bd test` on Linux x64:

- New or ported, failing without the change (24):
`serve-stream-body-error.test.ts` (8), `body-mixin-errors.test.ts` (9),
`bun-server.test.ts` (1), `fetch.stream.test.ts` (2),
`s3-stream-error-gc.test.ts` (2), `html-rewriter.test.js` (2). Each was
run against a debug build of main bd599f5, the first file against the
release canary.
- `s3-write-to-file-sync-close.test.ts` passes with and without the
change: main already rejects that write, and the test now pins it.
- After the rebase onto main (bd599f5): `serve-stream-body-error` 34
pass, `body-mixin-errors` 23, `s3-stream-error-gc` 4,
`s3-write-to-file-sync-close` 1, `html-rewriter.test.js` 188,
`bun-server` 81, `fetch.stream` 121.
- This container gives the debug build about two cores. `bun-server` and
`fetch.stream` were run with `--max-concurrency` lowered: with the
default, their concurrent subprocess tests pass the 5 s limit, with and
without the change. The "Content-Length response works (multiple parts)"
rows of `fetch.stream` take 2.1 s to 4.9 s each on both builds, and in
one run two of them passed the limit.
- Earlier, before the rebase: `fetch-backpressure` 127, `body-stream`
9086, `body` 774, `streams` 624, `node-stream` 112. `serve.test.ts`: 325
pass. "root range port" and "/bun:info to loopback" fail on the release
build of main too (the container runs as root).

`bytestream-partial-read-error-fixture.ts` stages the failure with an
`HTMLRewriter` whose handler throws on the second chunk, so every step
runs on the JS thread in order. It waits for the handler to have run,
checks that the first read is exactly one pull view, and requires the
next read to reject with the handler's own error. A text-mode reader is
required: a byte reader's full view retires the adapter's pending view,
and the next pull then skips `drain()`. Against the released bun the
fixture aborts with the panic in its header.

`s3-stream-stored-error-fixture.ts` has two modes. With no credentials
the download fails inside `stream()`, so every reader and sink runs
against a stored failure and no timing is involved. For the failures on
the wire, the stub closes the connection behind each answer. The read
starts after the client has closed its end and one more request has made
the round trip through the HTTP thread. The fixture writes to an
absolute path: `Bun.write("out.bin", new Response(""))` with a relative
path to a file that does not exist rejects with `ENOENT` from
`truncate`, on main too. That is a separate defect.

**Size.** `size` of the release binary: `text` 80,676,945 on main,
80,677,713 with the change. `data` and `bss` are equal. Measured before
the rebase.

**History.** The branch is rebased onto main at bd599f5: one commit for
the change, and one for the tests of the S3 and `HTMLRewriter` sources.

**Related open pulls.** #33461 and #36110 give opposite answers for a
started body that did not fail (send everything, or refuse). #42281
keeps `Readable.fromWeb()` from taking a started native source. None of
them touches `ByteStream.rs`. #43410 rewrites how `append(Err)` stores a
JS error and needs a rebase after this.

</details>

<!-- robobun:evidence:begin -->

---

**no test proof** · iteration 0 · platform-specific test(s) that do not
run on this machine, deferring to CI, which covers all platforms:
test/js/workerd/html-rewriter.test.js,
test/js/web/fetch/fetch.stream.test.ts,
test/js/bun/http/bun-server.test.ts

<!-- robobun:evidence:end -->

…axRequestBodySize

A chunked request body can cross maxRequestBodySize after the handler has
answered. The response cannot take its status back, so both over-limit arms
wrote a 413 that uWS drops and then ended the response as if it were
complete: the header lines of that second end landed in the body, and a
terminating chunk closed a body that was cut. A response whose body is a
native pipe kept its request and its upstream connection for ever.

One exit, `reject_oversized_request_body`, replaces both arms. It revokes the
body writer before it fails the request body, chooses the wire outcome by
phase (the 413 while no status is written, else the close of #40596), fires
`req.signal`, and tells the producer last.

`on_abort` revokes the writer before `detach_response()`, so a `req.signal`
listener that writes to a direct stream can no longer reach the response that
the server gave up.